package com.wzx.demo.conf;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * 消息队列业务具体逻辑实现
 *
 * @author wangzhengxing
 * @version 1.0
 * @Date 2019-10-31 09:00
 */
@Configuration
public class QueueMessageService implements QueueMessage {
    /**
     * 消息队列模板
     */
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void sendDeadMsg(String msg) throws Exception {
        /**
         * 设置回调为当前类对象
         */
        rabbitTemplate.setConfirmCallback(this::confirm);
        rabbitTemplate.setReturnCallback(this::returnedMessage);
        //当参数为true时，交换器无法根据自身的类型和路由键找到一个符合条件的队列，那么RabbitMQ会调用Return命令将消息返回给生产者；否则消息直接被丢弃
        rabbitTemplate.setMandatory(true);
        /**
         * 构建回调ID为UUID
         */
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        /**
         * 声明消息处理器
         */
        MessagePostProcessor messagePostProcessor = messageProcessor -> {
            MessageProperties messageProperties = messageProcessor.getMessageProperties();
            //设置编码
            messageProperties.setContentEncoding("utf-8");
            //设置过期时间,毫秒值
            messageProperties.setExpiration("2000");
            return messageProcessor;
        };
        /**
         * 发送消息到消息队列,10000毫秒后过期，成为死信
         */
        rabbitTemplate.convertAndSend(ExchangeEnum.DIRECT_DEAD_EXCHANGE.getExchangeName(), QueueEnum.DIRECT_DEAD_QUEUE.getRoutingKey(), msg, messagePostProcessor, correlationData);
    }

    public void sendDirectMsg(String msg) {
        rabbitTemplate.convertAndSend(ExchangeEnum.DIRECT_EXCHANGE.getExchangeName(), QueueEnum.DIRECT_QUEUE.getRoutingKey(), msg, new CorrelationData("2"));
        if (rabbitTemplate.waitForConfirms(1000)) {
            System.out.println("消息发送失败");
        }
    }

    public void sendFanoutMsg(String msg) {
        rabbitTemplate.convertAndSend(ExchangeEnum.FANOUT_EXCHANGE.getExchangeName(), "", msg, new CorrelationData("123213"));
    }

    public void sendTopicMsg(String msg) {
        rabbitTemplate.convertAndSend(ExchangeEnum.TOPIC_EXCHANGE.getExchangeName(), QueueEnum.TOPIC_QUEUE.getRoutingKey(), msg);
    }

    /**
     * 消息回调确认方法
     * @param correlationData   请求数据对象
     * @param ack               是否发送成功
     * @param cause             失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("回调id: " + correlationData.getId());
        if (ack) {
            System.out.println("消息发送成功");
        } else {
            System.out.println("消息发送失败： " + cause);
        }
    }

    /**
     * 当消息从交换机到队列失败时，该方法被调用。若成功，则不调用。注意：当方法调用后，confirm方法也会被调用，且ack为true
     * @param message       发送的消息
     * @param replyCode     返回码
     * @param replyText     返回消息
     * @param exchange      交换机
     * @param routingKey    路由键
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("message: " + message + "; replyCode: " + replyCode + "; exchange: " + exchange + "; routingKey: " + routingKey);
    }
}
